package com.atguigu.edu.realtime.app.dwd.db;

import com.atguigu.edu.realtime.util.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @ClassName: DwdTradePaySuccess
 * @Author: ghy
 * @Date: 2022/10/19 23:42
 * @Description:
 **/
public class DwdTradePaySuccess {
    public static void main(String[] args) {
        //环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(15*60 + 10));
        //检查点
        //2.1 开启检查点
//        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
//        //2.2 设置检查点超时时间
//        env.getCheckpointConfig().setCheckpointTimeout(60000L);
//        //2.3 设置job取消之后，检查点是否保留
//        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//        //2.4 设置两个检查点之间最小时间间隔
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
//        //2.5 设置重启策略
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30),Time.seconds(3)));
//        //2.6 设置状态后端
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop101:8020/gmall/ck");
//        //2.7 设置操作hadoop的用户
//        System.setProperty("HADOOP_USER_NAME","atguigu");
        //3.从kafka的topic_db主题中读取数据 创建动态表
        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL("dwd_trade_pay_success_group"));

        //4.过滤支付表数据
        Table paymentInfo = tableEnv.sqlQuery("select \n" +
                "data['id'] id,\n" +
                "data['order_id'] order_id,\n" +
                "data['total_amount'] total_amount,\n" +
                "data['payment_status'] payment_status,\n" +
                "data['callback_time'] callback_time,\n" +
                "ts \n" +
                "from `topic_db`\n" +
                "where `table` = 'payment_info'\n" );
        tableEnv.createTemporaryView("payment_info",paymentInfo);


        //5.过滤订单明细数据
        Table orderDetail = tableEnv.sqlQuery("select \n" +
                "data['id'] id,\n" +
                "data['course_id'] course_id,\n" +
                "data['course_name'] course_name,\n" +
                "data['order_id'] order_id,\n" +
                "data['user_id'] user_id,\n" +
                "data['origin_amount'] origin_amount,\n" +
                "data['coupon_reduce'] coupon_reduce,\n" +
                "data['final_amount'] final_amount,\n" +
                "data['session_id'] session_id,\n" +
                "`type`,\n" +
                "`old`,\n" +
                "ts od_ts\n" +
                "from `topic_db`\n" +
                "where `table` = 'order_detail'\n" +
                "and (`type` = 'insert' or `type` = 'update')");
        tableEnv.createTemporaryView("order_detail", orderDetail);
        //tableEnv.executeSql("select * from order_detail").print();


        //6.关联
        Table resultTable = tableEnv.sqlQuery("select \n" +
                "pi.id,\n" +
                "pi.order_id,\n" +
                "pi.total_amount,\n" +
                "pi.payment_status,\n" +
                "date_format(pi.callback_time, 'yyyy-MM-dd') date_id,\n" +
                "pi.callback_time,\n" +
                "pi.ts,\n" +

                "od.course_id,\n" +
                "od.course_name,\n" +
                "od.user_id,\n" +
                "od.origin_amount,\n" +
                "od.coupon_reduce,\n" +
                "od.final_amount,\n" +

                "current_row_timestamp() row_op_ts\n" +

                "from payment_info pi\n" +
                "join order_detail od\n" +
                "on pi.order_id = od.order_id\n" +
                "where pi.payment_status = '1602'"
        );
        tableEnv.createTemporaryView("result_table", resultTable);
        // tableEnv.executeSql("select * from result_table").print();

        //5.创建动态表，映射
        tableEnv.executeSql("create table dwd_trade_pay_success(\n" +
                "id string,\n" +
                "order_id string,\n" +
                "total_amount string,\n" +
                "payment_status string,\n" +
                "date_id string,\n" +
                "callback_time string,\n" +
                "ts string,\n"  +

                "course_id string,\n" +
                "course_name string,\n" +
                "user_id string,\n" +
                "origin_amount string,\n" +
                "coupon_reduce string,\n" +
                "final_amount string,\n" +

                "row_op_ts timestamp_ltz(3),\n" +
                "primary key(id) not enforced\n" +
                ")" + MyKafkaUtil.getUpsertKafkaDDL("dwd_trade_pay_success"));

        //tableEnv.executeSql("select * from dwd_trade_pay_success").print();
        //6.写入
        tableEnv.executeSql("insert into dwd_trade_pay_success select * from result_table");
    }
}
